package org.rency.mq.kafka.consumer.support.factory;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.internals.NoOpConsumerRebalanceListener;
import org.apache.kafka.common.serialization.Deserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;

/**
 * Kafka消费者实现类
 *
 * @author: Created by user_rcy@163.com on 2016/10/21.
 * @version: V1.0.0
 * @since JDK 1.8
 */
@Deprecated
public class DefaultKafkaConsumer<K, V> implements FactoryBean<Consumer<K,V>>,InitializingBean{

    private static final Logger logger = LoggerFactory.getLogger(DefaultKafkaConsumer.class);

    private final String VALIDATION_ERROR = "Kafka Consumer Configuration Validation Error, '%s' is empty.";

    @Resource(name="consumerProperties")
    private Map<String, Object>                   consumerProperties;

    private Deserializer<K>                       keyDeserializer;

    private Deserializer<V>                       valueDeserializer;

    private List<String>                          topics;

    private ConsumerRebalanceListener             rebalanceListener = new NoOpConsumerRebalanceListener();

    private KafkaConsumer<K, V> getInstance(){
        KafkaConsumer<K, V> kafkaConsumer = new KafkaConsumer<K, V>(this.consumerProperties,this.keyDeserializer,this.valueDeserializer);
        kafkaConsumer.subscribe(this.topics,this.rebalanceListener);
        return kafkaConsumer;
    }

    @Override
    public Consumer<K, V> getObject() throws Exception {
        return this.getInstance();
    }

    @Override
    public Class<?> getObjectType() {
        return Consumer.class;
    }

    @Override
    public boolean isSingleton() {
        return false;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        logger.info("Kafka Consumer Configuration Validation.");
        try {
            Assert.notEmpty(this.consumerProperties,String.format(VALIDATION_ERROR,"consumerProperties"));
            Assert.notNull(this.keyDeserializer,String.format(VALIDATION_ERROR,"keyDeserializer"));
            Assert.notNull(this.valueDeserializer,String.format(VALIDATION_ERROR,"valueDeserializer"));
            Assert.notEmpty(this.topics,String.format(VALIDATION_ERROR,"topics"));
        }catch (IllegalArgumentException e){
            logger.error("Kafka Consumer Configuration Validation Failed.",e);
            throw new RuntimeException(e);
        }
    }

    public void setKeyDeserializer(Deserializer<K> keyDeserializer) {
        this.keyDeserializer = keyDeserializer;
    }

    public void setValueDeserializer(Deserializer<V> valueDeserializer) {
        this.valueDeserializer = valueDeserializer;
    }

    public void setTopics(List<String> topics) {
        this.topics = topics;
    }

    public void setRebalanceListener(ConsumerRebalanceListener rebalanceListener) {
        this.rebalanceListener = rebalanceListener;
    }
}
